package studio.raptor.extractor.segment.config;

import com.zaxxer.hikari.HikariDataSource;
import org.apache.zookeeper.server.ZKDatabase;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.jdbc.DataSourceBuilder;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import studio.raptor.extractor.segment.entity.BuildJsonResult;
import studio.raptor.extractor.segment.entity.EventMeta;
import studio.raptor.extractor.segment.properties.MQProperties;
import studio.raptor.extractor.segment.properties.QueueProperties;
import studio.raptor.extractor.segment.properties.ThreadProperties;
import studio.raptor.extractor.segment.properties.ZKProperties;
import studio.raptor.gid.Sequencer;
import studio.raptor.gid.def.BreadcrumbDef;


import javax.sql.DataSource;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * Created by liuyong8 on 2017/7/17.
 */
@Configuration
public class ExtractorConfig {

  @Bean(name = "extractorDataSource")
  @ConfigurationProperties(prefix = "extractor.datasource")
  public DataSource extractorDataSource() {
    return DataSourceBuilder.create().type(HikariDataSource.class).build();
  }

  @Bean
  @ConfigurationProperties(prefix = "extractor.mq")
  public MQProperties mqProperties() {
    return new MQProperties();
  }

  @Bean
  public ZKProperties zkProperties() {
    return new ZKProperties();
  }

  @Bean
  @ConfigurationProperties(prefix = "extractor.queue")
  public QueueProperties queueProperties() {
    return new QueueProperties();
  }

  @Bean
  @ConfigurationProperties(prefix = "extractor.thread")
  public ThreadProperties threadProperties() {
    return new ThreadProperties();
  }

  @Bean(destroyMethod = "shutdown")
  @ConditionalOnBean(ZKProperties.class)
  public Sequencer sequencer(ZKProperties zkProperties) throws Exception {
    Sequencer server = new Sequencer(zkProperties.getServers(), zkProperties.getNamespace(), "");
    server.startup();
    server.add(new BreadcrumbDef() {

      @Override
      public long start() {
        return 100000000000000000l;
      }

      @Override
      public String name() {
        return "metaId";
      }

      @Override
      public long incr() {
        return 1;
      }

      @Override
      public int cache() {
        return 1000;
      }
    });
    return server;
  }
}
